package com.zx.mes.coolmq.hyl.config;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zx.mes.coolmq.hyl.util.MqConstants;
import com.zx.mes.coolmq.hyl.util.RabbitMetaMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;

/**
 * ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调。
 * ReturnCallback接口用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调
 * RabbitTemplate配置类
 * @author hyl
 * @date 2018-3-19
 * @version 1.0.0
 */
@Configuration
public class RabbitTemplateConfig {

    private static final Logger logger = LoggerFactory.getLogger(RabbitTemplateConfig.class);

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    private ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 定制化amqp模板 可以根据需要定制多个
     * 此处为模板类定义 Jackson消息转换器
     * confirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调 即消息发送到exchange ack
     * ReturnCallback接口用于实现消息发送到RabbitMQ交换器，但无相应队列与交换机绑定时的回调，即消息发送不到任何的
     * @return the amqp(rabbitMQ) template
    */
    @Bean
    public RabbitTemplate rabbitTemplate(){
//        使用jackson消息转换器
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        rabbitTemplate.setEncoding("UTF-8");
//        开启returnCallback yml需要配置 publisher-returns:true
        rabbitTemplate.setMandatory(true);

        //消息发送到RabbitMQ交换器，但无相应Exchange时的回调
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                String cacheKey = message.getMessageProperties().getMessageId();
                logger.error("returnCallback,没有找到任何匹配的队列!message id:{},replyCode:{},replyText:{},exchange:{},routingKey:{}",
                        cacheKey,replyCode,replyText,exchange,routingKey);
//                String KEYS="HASH:";rt.opsForHash().put(KEYS, "name", "zhang");System.out.println(rt.opsForHash().get(KEYS, "name"));
//                RabbitMetaMessage metaMessage = (RabbitMetaMessage) redisTemplate.opsForHash().get(MqConstants.MQ_PRODUCER_RETRY_KEY, cacheKey);
                ValueOperations<String, RabbitMetaMessage> operations=redisTemplate.opsForValue();
                RabbitMetaMessage metaMessage=operations.get(cacheKey);
                metaMessage.setReturnCallback(true);
//                由于amqp奇葩协议规定，return比confirm先回调
//                redisTemplate.opsForHash().put(MqConstants.MQ_PRODUCER_RETRY_KEY,cacheKey,message);
                operations.set(cacheKey,metaMessage);
            }
        });
//        消息发送到RabbitMQ交换器后接收ack回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                logger.debug("confirmCallback,CorrelationData:{},ack:{},cause:{}", correlationData, ack, cause);
                String cacheKey = correlationData.getId();
//                RabbitMetaMessage metaMessage = (RabbitMetaMessage) redisTemplate.opsForHash().get(MqConstants.MQ_PRODUCER_RETRY_KEY, cacheKey);
                ValueOperations<String, RabbitMetaMessage> operations=redisTemplate.opsForValue();
                RabbitMetaMessage metaMessage=operations.get(cacheKey);
//                只要消息能投入正确的交换器中，并持久化，就返回ack为true
                if (ack) {
                    if (!metaMessage.isReturnCallback()) {
                        logger.info("消息已正确投递到队列,correlationData:{}",correlationData);
//                        清除缓存
//                        redisTemplate.opsForHash().delete(MqConstants.MQ_PRODUCER_RETRY_KEY, cacheKey);
                        stringRedisTemplate.delete(cacheKey);
                    }

                } else {
                    logger.warn("交换机投机消息至队列失败，correlationData:{},原因:{}", correlationData,cause);
                    if (!metaMessage.isAutoTrigger()) {
                        metaMessage.setAutoTrigger(true);
                        try {
//                            redisTemplate.opsForHash().put(MqConstants.MQ_PRODUCER_RETRY_KEY,cacheKey,objectMapper.writeValueAsString(metaMessage));
                            operations.set(cacheKey,metaMessage);
                        } catch (Exception e) {
                            logger.error("删除redis重发缓存失败,e:{}",e);
                        }
                    }
                }
           }
        });
        return rabbitTemplate;
    }
}
